<?php
namespace database;

use config\Config;

class MyKafka{
    private static $kafkaProduceInstance;
    private static $kafkaConsumerInstance;

    private function __construct(){

    }

    //kafka生产连接
    public static function getProduceConn(){
        if(!self::$kafkaProduceInstance instanceof self){
            self::$kafkaProduceInstance = new self;
        }

        $temp = self::$kafkaProduceInstance;
        return $temp->connectKafkaProduce();

    }

    //kafka消费连接
    public static function getConsumerConn(){
        if(!self::$kafkaConsumerInstance instanceof self){
            self::$kafkaConsumerInstance = new self;
        }

        $temp = self::$kafkaConsumerInstance;
        return $temp->connectKafkaConsumer();

    }

    private static function connectKafkaProduce(){
        $kfkConf = Config::config("kafka");
        try{
            $produce = \Kafka\Produce::getInstance($kfkConf["host"],$kfkConf["timeout"]);
            $produce->setRequireAck(-1);
        }catch (\Exception $e){
            $errRs = array(
                "errMsg" => "kafka produce connect err",
                "msg"=>$e->getMessage(),
            );
            mLogs("error","kafka_connect",$errRs,"kafka");
        }
        return $produce;

    }

    private static function connectKafkaConsumer(){
        $kfkConf = Config::config("kafka");
        try{
            $consumer = \Kafka\Consumer::getInstance($kfkConf["host"]);
        }catch (\Exception $e){
            $errRs = array(
                "errMsg" => "kafka consumer connect err",
                "msg"=>$e->getMessage()
            );
            mLogs("error","kafka_connect",$errRs,"kafka");
        }
        return $consumer;
    }

}